package com.soft863.ct.consumer.dao;

import com.soft863.ct.consumer.bean.CallLog;
import com.soft863.tc.common.bean.BaseDao;
import com.soft863.tc.common.constant.Names;
import com.soft863.tc.common.constant.ValueConstant;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;

/*
 * HBase的数据访问对象
 * */
public class HBaseDao extends BaseDao {
    /*
     * 初始化
     * */
    public void init() throws IOException {
        start();
        creatNamespaceNX(Names.NAMESPACE.getValue());
        createTableXX(Names.TABLE.getValue(),
                "com.soft863.tc.consumer.copeocessor.InsertCalleeCoprocessor",
                ValueConstant.REGION_COUNT,
                Names.CF_CALLER.getValue(),
                Names.CF_CALLEE.getValue());
        end();
    }

    /**
     * 将对象写入HBase
     *
     * @param log obj
     * @throws IOException            '
     * @throws IllegalAccessException '
     */
    public void insertData(CallLog log) throws IOException, IllegalAccessException {
        log.setRowKey(genRegionNum(log.getCalling(), log.getCallTime()) + "_" +
                log.getCalling() + "_" +
                log.getCallTime() + "_" +
                log.getCalled() + "_" +
                log.getDuration());

        putData(log);
    }

    /**
     * 将字符串写入HBase，该方法使用不友好
     *
     * @param value '
     * @throws IOException '
     */
    public void insertData(String value) throws IOException {
        //将通话日志保存到HBase表中
        //获取通话日志数据
        String[] values = value.split("\t");
        String calling = values[0];
        String called = values[1];
        String callTime = values[2];
        String duration = values[3];
        /*
         * rowKey设计
         * 1）长度原则 => 最大值64kb，推荐长度 10 ~ 100 byte，最好是8的倍数，能短则短，影响存储性能
         * 2）唯一原则 => rowKey 唯一
         * 3）散列原则 => 盐值散列（不能使用时间戳）、字符串翻转（如翻转时间戳，电话号码）、计算分区号
         * */
        //rowKey = regionNum + calling + time + called + duration
        //主叫 _1
        String rowKey = genRegionNum(calling, callTime) + "_" + calling + "_" + callTime + "_" + called + "_" + duration + "_1";
        //todo : 有bug 有默认的列族名CF_DEFAULT
        byte[] family = Bytes.toBytes(Names.CF_CALLER.getValue());
        //创建数据对象
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(family, Bytes.toBytes("call1"), Bytes.toBytes(calling));
        put.addColumn(family, Bytes.toBytes("call2"), Bytes.toBytes(called));
        put.addColumn(family, Bytes.toBytes("callTime"), Bytes.toBytes(callTime));
        put.addColumn(family, Bytes.toBytes("duration"), Bytes.toBytes(duration));
        put.addColumn(family, Bytes.toBytes("flag"), Bytes.toBytes("1"));

        //被叫 _0
        //考虑到 1 -> 2 通话，查询2的通话记录这条数据也应该显示，因此在插入1 -> 2 这条数据时，同时添加 2 —> 1 数据
        //为了提高查询效率，增加列族信息，列族分为主叫和被叫，因为列族是分文件夹存储，查询效率高
        //使用协处理插入数据
//        String rowKey1 = genRegionNum(called, callTime) + "_" + called + "_" + callTime + "_" + calling + "_" + duration + "_0";
//        byte[] family1 = Bytes.toBytes(Names.CF_CALLEE.getValue());
//        Put put1 = new Put(Bytes.toBytes(rowKey1));
//        put1.addColumn(family1, Bytes.toBytes("call1"), Bytes.toBytes(called));
//        put1.addColumn(family1, Bytes.toBytes("call2"), Bytes.toBytes(calling));
//        put1.addColumn(family1, Bytes.toBytes("callTime"), Bytes.toBytes(callTime));
//        put1.addColumn(family1, Bytes.toBytes("duration"), Bytes.toBytes(duration));
//        put1.addColumn(family1, Bytes.toBytes("flag"), Bytes.toBytes("0"));

        ArrayList<Put> puts = new ArrayList<>();
        puts.add(put);
//        puts.add(put1);
        putData(Names.TABLE.getValue(), puts);
    }

}
